package cn.lyjuan.first.netty.demo._2_netty.selector;

import cn.lyjuan.first.netty.demo._2_netty.NIOPool;
import com.sun.org.apache.bcel.internal.generic.Select;

import java.io.IOException;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 抽象Selector，Boss和Worker的Selector共有操作
 */
public abstract class AbstractNIOSelector implements Runnable
{
    /**
     * 选择器
     */
    private Selector selector;

    /**
     * 唤醒状态标识
     */
    private AtomicBoolean wakeFlag = new AtomicBoolean(true);

    /**
     * 注册任务队列
     */
    private final Queue<Runnable> queue = new ConcurrentLinkedQueue<Runnable>();

    /**
     * 线程名称
     */
    private String threadName;

    /**
     * 线程池
     */
    protected Executor executor;

    /**
     * 线程池管理对象
     */
    protected NIOPool pool;

    public AbstractNIOSelector init(String threadName, Executor executor, NIOPool pool)
    {
        this.threadName = threadName;
        this.executor = executor;
        this.pool = pool;

        openSelector();

        return this;
    }

    private void openSelector()
    {
        try
        {
            this.selector = Selector.open();
        } catch (IOException e)
        {
            throw new RuntimeException(e);
        }

        // 线程池中执行
        this.executor.execute(this);
    }

    /**
     * 线程化运行
     */
    public void run()
    {
        Thread.currentThread().setName(threadName);

        while (true)
        {
            // 设置在处理中标识，需唤醒
            wakeFlag.set(false);

            // 监听选择事件
            select();

            // 处理新增监听任务
            processQueue();

            // 处理核心任务
            processKeys();
        }
    }

    /**
     * 监听，并处理监听到的事件
     */
    private int select()
    {
        int count = 0;

        try
        {
            count = this.selector.select(500);
        } catch (IOException e)
        {
            e.printStackTrace();
        }

        return count;
    }

    /**
     * 处理任务队列
     */
    private void processQueue()
    {
        while (true)
        {
            Runnable run = queue.poll();

            if (null == run) break;

            // 在线程池中执行任务
            run.run();
        }
    }

    /**
     * 处理所监听到的事件
     */
    private void processKeys()
    {
        Iterator<SelectionKey> it = this.selector.selectedKeys().iterator();
        while (it.hasNext())
        {
            SelectionKey key = it.next();
            it.remove();
            process(key);
        }
    }

    /**
     * 处理核心事件
     */
    protected abstract void process(SelectionKey selectionKey);

    /**
     * 监听类型
     *
     * @return
     */
    protected abstract Integer[] selectionKeys();

    /**
     * 注册监听选择
     */
    public void register(final SelectableChannel chl)
    {
        final Integer[] selectionKeys = selectionKeys();

        try
        {
            chl.configureBlocking(false);
        } catch (IOException e)
        {
            e.printStackTrace();
        }

        // 将注册事件添加到任务队列中
        queue.add(new Runnable()
        {
            public void run()
            {
                int key = 0;

                for (Integer k : selectionKeys)
                    key |= k;

                try
                {
                    chl.register(AbstractNIOSelector.this.selector, key);
                } catch (ClosedChannelException e)
                {
                    throw new RuntimeException(e);
                }
            }
        });

        // 如有必要，唤醒 selector 监听
        boolean waking = wakeFlag.compareAndSet(false, true);
        if (waking)
        {
            this.selector.wakeup();
        }
    }
}
